package baiduschPro;

import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.flowfile.FlowFile;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.io.StreamCallback;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.BufferedInputStream;
import org.apache.nifi.stream.io.BufferedOutputStream;
import org.apache.nifi.stream.io.ByteArrayOutputStream;

import baidusch.BaiduSchResultParse;
import hylanda.common.FileWr;

@Tags({"single baidu-search;input "})
@CapabilityDescription("baidu search genarate url")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})

public class BaiduSchParseProcessor extends AbstractProcessor {

	
	public static final PropertyDescriptor MY_PROPERTY = new PropertyDescriptor
            .Builder().name("MY_PROPERTY")
            .displayName("regRule Center")
            .description("正则文件获取中心的地址")
            .required(false)
           // .defaultValue("")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();
	
	 public static final Relationship MY_RELATIONSHIPOK = new Relationship.Builder()
	            .name("gen success")
	            .description("success relationship")
	            .build();
	 
	 private List<PropertyDescriptor> descriptors;

	 private Set<Relationship> relationships;
	 
	public BaiduSchParseProcessor() {
		// TODO Auto-generated constructor stub
	}
	
	 @Override
	    public Set<Relationship> getRelationships() {
	        return this.relationships;
	    }

	    @Override
	    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
	        return descriptors;
	    }
	
	 @Override
	    protected void init(final ProcessorInitializationContext context) {
	        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
	        descriptors.add(MY_PROPERTY);
	      
	        this.descriptors = Collections.unmodifiableList(descriptors);
	        final Set<Relationship> relationships = new HashSet<Relationship>();
	        relationships.add(MY_RELATIONSHIPOK);	      
	        this.relationships = Collections.unmodifiableSet(relationships);
	        
	        
	    }

	
  @Override
  public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
	  
		  FlowFile flowFile = session.get();
	      if ( flowFile == null ) {
	      	
	      	System.out.println("flowfile is null");
	         return;
	      }
	      
	      //读取FlowFile
	        String content = "";
	        InputStream in = session.read(flowFile);
	        try {
				byte[] by = readStream(in);
				content = new String(by,"UTF-8");
				
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				//return ;
			}
	        
	        //获取Property
//	       PropertyValue myproptyp = context.getProperty("MY_PROPERTY");
//		   String centerurl = myproptyp.getValue();
		   
		   //具体业务---解析正文
		   String result =BaiduSchResultParse.parsePage(content);

// FileWr.saveFile("e:\\code\\baiduschtestresultnew.json", result);
		   
		   //写flowFile，提交
		   flowFile= session.write(flowFile, new OutputStreamCallback(){
	    		@Override
	    		public void process(OutputStream out) throws IOException{
	    			out.write(result.getBytes("UTF-8"));
	    		}
	    	}); 
	        session.transfer(flowFile,MY_RELATIONSHIPOK);
	       
  }
	
  
  public static String getCurrentTime()
	{
		SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
		String str = df.format(new Date());// new Date()为获取当前系统时间
		return str;
	}
	 public static byte[] readStream(InputStream inStream) throws Exception {  
	        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();  
	        byte[] buffer = new byte[1024];  
	        int len = -1;  
	        while ((len = inStream.read(buffer)) != -1) {  
	            outSteam.write(buffer, 0, len);  
	        }  
	        outSteam.close();  
	        inStream.close();  
	        return outSteam.toByteArray();  
	    }  
	

}
